package io.confluent.examples.producer

import java.util.{Date, Properties, Random}

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

/**
  * Created by hunter on 17-11-19.
  */
class ProducerExample {


}

/**
  * ./bin/kafka-avro-console-consumer  --bootstrap-server localhost:9092 --from-beginning --topic page_visits --property value.schema='{"namespace": "example.avro", "type": "record","name": "page_visit","fields": [{"name": "time", "type": "long"},{"name": "site", "type": "string"},{"name": "ip", "type": "string"}]}'
  */
object ProducerExample{
  val EVENT_COUNT = 10
  val SCHEMA_REGISTRY_URL = "http://localhost:8081"
  def main(args:Array[String]):Unit = {
    val props:Properties = new Properties()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092")
    props.put(ProducerConfig.ACKS_CONFIG,"all")
    props.put(ProducerConfig.RETRIES_CONFIG,new Integer(0))
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"io.confluent.kafka.serializers.KafkaAvroSerializer")
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"io.confluent.kafka.serializers.KafkaAvroSerializer")
    props.put("schema.registry.url",SCHEMA_REGISTRY_URL)


    val schemaString:String =  """{"namespace": "example.avro", "type": "record",
      |"name": "page_visit",
      |"fields": [
      |{"name": "time", "type": "long"},
      |{"name": "site", "type": "string"},
      |{"name": "ip", "type": "string"}
      |]}""".stripMargin

    val producer:KafkaProducer[String,GenericRecord] = new KafkaProducer[String,GenericRecord](props)

    val parser:Schema.Parser = new Schema.Parser()
    val schema:Schema = parser.parse(schemaString)


    val rnd:Random = new Random()
    for (i <- 0 until EVENT_COUNT){
      val runtime = new Date().getTime
      val site = "www.example.com"
      val ip = s"192.168.2.${rnd.nextInt(255)}"

      val page_visit = new GenericData.Record(schema)
      page_visit.put("time",runtime)
      page_visit.put("site",site)
      page_visit.put("ip",ip)


      val data= new ProducerRecord[String,GenericRecord]("page_visits",ip,page_visit)
      producer.send(data)
    }
    producer.close()
  }
}
